/**
 * Copyright (c) 2011-2015 1lifes.com
 * HSJ java-handlersocket
 * Id: HSJClient.java
 */
package com.elifes.hsj.client;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import com.elifes.hsj.IDBConfigLoader;
import com.elifes.hsj.client.netty.HSClientPipelineFactory;
import com.elifes.hsj.exception.HSJException;
import com.elifes.hsj.exception.HSJTimeoutException;
import com.elifes.hsj.model.DBConfig;
import com.elifes.hsj.packet.IPacket;
import com.elifes.hsj.packet.InsertPacket;
import com.elifes.hsj.packet.OpenIndexPacket;
import com.elifes.hsj.packet.PacketFactory;
import com.elifes.hsj.util.ChannelItem;
import com.elifes.hsj.util.IDGenerator;

/**
 * 描述：handlersocket 通讯（可支持分库分表）
 * @author yangqiang
 * @createtime 2011-11-12下午12:10:15
 *
 */
public class HSJClient{
	private static Logger logger = Logger.getLogger(HSJClient.class);
	private ClientBootstrap bootstrap;
	private IDBConfigLoader dbConfigLoader;
	private String authKey;
	
	/** <channelId, packet> */
	private ConcurrentHashMap<Integer,IPacket> pendingRequest = new ConcurrentHashMap<Integer,IPacket>();

	/** <indexId, dbName> 用于获取对应的连接 */
	private ConcurrentHashMap<String, String> indexIdDbNameMap = new ConcurrentHashMap<String, String>();
	
	/** 用于动态添加连接时进行openindex操作 */
	private ConcurrentHashMap<String, OpenIndexPacket> indexIdOpenIndexMap = new ConcurrentHashMap<String, OpenIndexPacket>();

	/** 插入失败 重试队列 */
	private BlockingQueue<InsertPacket> retryInsertQueue = new LinkedBlockingQueue<InsertPacket>(1000);
	
	private Map<String, InetSocketAddress> dbHostMap = new HashMap<String, InetSocketAddress>();
	
	private Lock timeLock = new ReentrantLock();
	private Condition timeCondition = timeLock.newCondition();
	
	private Lock connLock = new ReentrantLock();	
	
	private ExecutorService dbOperationExecutor = Executors.newFixedThreadPool(80);
	
	private ExecutorService connExecutor = Executors.newFixedThreadPool(10);
	
	/** <dbName, n channels> */
	private Map<String, LinkedBlockingQueue<ChannelItem>> channelPool = new ConcurrentHashMap<String, LinkedBlockingQueue<ChannelItem>>();
	/** 用于连接池超时检查 */
	private Map<ChannelItem, Boolean> channelItemMap = new ConcurrentHashMap<ChannelItem, Boolean>();
	private int poolSize = 20;
	private int maxActive = 100;
	private int idleTime = 10 * 60 * 60;
	private AtomicInteger curConnCount = new AtomicInteger();
	private int timeout = 2000;

	
	private static HSJClient self = new HSJClient();

	public HSJClient(){
		bootstrap = new ClientBootstrap(
				new NioClientSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));
		bootstrap.setPipelineFactory(new HSClientPipelineFactory(this));
		bootstrap.setOption("tcpNoDelay", true);
	}
	
	//after constructor,only invoke once
	public void init() throws HSJTimeoutException{
		List<DBConfig> dbConfigList = dbConfigLoader.load();
		
		InetSocketAddress address = null;
		for(DBConfig dbConfig : dbConfigList){
			authKey = dbConfig.getAuthKey();
			address = new InetSocketAddress(dbConfig.getIp(),
					dbConfig.getPort());
			dbHostMap.put(dbConfig.getDbName(), address);
			
			//初始化 就建立多个channel 
			initConnectionPool(address, dbConfig.getDbName());
		}
		
		Thread retryInsertThread = new Thread() {
            public void run() {
            	while(true){
            		InsertPacket insertPacket = null;
                    	try {
							insertPacket = retryInsertQueue.take();
							innerInsert(insertPacket);
						} catch (InterruptedException e) {
							logger.error("retry insert InterruptedException!");
							retryInsertQueue.offer(insertPacket);
						} catch (HSJException e) {
							logger.error("retry insert error! " + e.getMessage());
							retryInsertQueue.offer(insertPacket);
						}
                    	
                                		
            	}
            }
		};
		
		Thread idleConnCheckThread = new Thread() {
            public void run() {
            	List<ChannelItem> removeList = new ArrayList<ChannelItem>();
            	while(true){
            		removeList.clear();
            		timeLock.lock();
	            	try {
	            		timeCondition.await(1, TimeUnit.MINUTES);
	            		if(channelItemMap.size() > poolSize){//当前连接数大于连接池初始数，超时回收
							for(final Entry<ChannelItem, Boolean> entry : channelItemMap.entrySet()){
								if(System.currentTimeMillis() - entry.getKey().getUseTime() >= idleTime){
									ChannelFuture future = entry.getKey().getItem().close();
									future.addListener(new ChannelFutureListener() {
										public void operationComplete(ChannelFuture future) throws Exception {
											if(future.isSuccess()){
												entry.getKey().setItem(null);
											}
										}
									});
								}
								if(entry.getKey().getItem() == null){
									removeList.add(entry.getKey());
								}
							}
							
							//real remove
							for(ChannelItem channelItem : removeList){
								channelItemMap.remove(channelItem);
							}
	            		}
					} catch (InterruptedException e) {
						logger.error("idleConnCheckThread check error!", e);
					}finally{
						timeLock.unlock();
					}
	            	
	            }
            }
		};
		
		idleConnCheckThread.setDaemon(true);
		retryInsertThread.setDaemon(true);
		idleConnCheckThread.start();
		retryInsertThread.start();
	}
	
	private void initConnectionPool(InetSocketAddress address, String dbName) throws HSJTimeoutException {
		LinkedBlockingQueue<ChannelItem> channelQueue = new LinkedBlockingQueue<ChannelItem>();
		ChannelItem channelItem = null;
		for(int i = 0; i < poolSize; i++){
			curConnCount.incrementAndGet();
			channelItem = addConn(address, dbName, null);
			if(channelItem != null){
				channelQueue.offer(channelItem);
				channelItemMap.put(channelItem, true);
			}
		}
		channelPool.put(dbName, channelQueue);
	}
	
	/**
	 * 
	 * 描述：和hs服务器进行网络连接
	 * 
	 * @param address 数据库服务器地址
	 * @param dbName 数据库名称
	 * @param indexId openIndex id 为空是初始化连接；非空 动态添加，需要进行openindex操作
	 * @throws HSJTimeoutException 
	 */
	private ChannelItem addConn(InetSocketAddress address, String dbName, String indexId) throws HSJTimeoutException{
		ChannelFuture future = bootstrap.connect(address);
		Channel channel = getChannel(future);
		if(channel != null){
			ChannelItem channelItem = new ChannelItem(System.currentTimeMillis(), channel);
			
			if(innerAuth(channelItem, null, authKey)){
				channelItem.setAuth(true);
				
				if(indexId != null && !innerOpenIndex(channelItem, indexIdOpenIndexMap.get(indexId))){//could't openIndex
					logger.warn("dynamic adjust add conn error, could't openIndex!");
					curConnCount.decrementAndGet();
					channelItem = null;
					
					return null;
				}
			} else {
				logger.warn("dynamic adjust add conn error, could't auth!");
				curConnCount.decrementAndGet();
				channelItem = null;
				
				return null;
			}
			
			return channelItem;
		} else {
			logger.warn("add conn error!");
			curConnCount.decrementAndGet();
		}
		
		return null;
	}
	
	public static HSJClient getInstance(){
		return self;
	}
	
	public void setDbConfigLoader(IDBConfigLoader dbConfigLoader){
		this.dbConfigLoader = dbConfigLoader;
	}
	
	/**
	 * 描述：先检查
	 * @param dbName
	 * @param tblName
	 * @param indexName
	 * @param columnNames
	 * @return
	 * @throws HSJException
	 */
	public String requsetOpenIndex(String oldIndexId, String dbName, String tblName,
			String indexName, List<String> columnNames, List<String> filterColumnNames) throws HSJException {
		String indexId = null;
		//支持重复openindex
		if(oldIndexId != null && oldIndexId.trim().length() != 0){
			indexId = oldIndexId;
		} else {
			indexId = String.valueOf(IDGenerator.nextId());
		}

		try {
			LinkedBlockingQueue<ChannelItem> dbQueue = channelPool.get(dbName);
			for(int i = 0, size = dbQueue.size(); i < size; i++){
				ChannelItem channelItem = dbQueue.take();
				
				if(channelItem.isAuth()){// has authed
					if(logger.isDebugEnabled()){
						logger.debug("start open index : " + dbName);
					}
					if(innerOpenIndex(channelItem, indexId, dbName, tblName,
							indexName, columnNames, filterColumnNames)){
						//indexIdDbNameMap.putIfAbsent(indexId, dbName);
					} else {
						//失败,从连接池移除channel并关闭
						logger.warn("openindex error!");
						handleOpenIndexFailed(channelItem, indexId);
						//throw new HSJException("openindex error!");
					}
					
					//break;
				} else {// reauth
					logger.warn("the channel not authed! id is: " + channelItem.getItem().getId());
					if(innerAuth(channelItem, null, authKey)){
						channelItem.setAuth(true);
						
						if(innerOpenIndex(channelItem, indexId, dbName, tblName,
								indexName, columnNames, filterColumnNames)){
							//indexIdDbNameMap.putIfAbsent(indexId, dbName);
						} else {
							//throw new HSJException("after reauth openindex error!");
							logger.warn("after reauth openindex error!");
							handleOpenIndexFailed(channelItem, indexId);
						}
					} else {
						//throw new HSJException("reauth error!");
						logger.warn("reauth error!");
						handleOpenIndexFailed(channelItem, indexId);
					}
				}
			}

		} catch (InterruptedException e) {
			throw new HSJException("dbQueue take error!", e);
		}
		
		return indexId;
	}
	
	private void handleOpenIndexFailed(ChannelItem channelItem, String indexId){
		//TODO handle close failed
		channelItem.getItem().close();
		channelItemMap.remove(channelItem);
		indexIdDbNameMap.remove(indexId);
		channelItem = null;
	}

	

	public boolean requsetInsert(IPacket insertPacket) throws HSJException {
		return (Boolean) sendRequest(insertPacket, false);
	}
	
	private void innerInsert(IPacket insertPacket) throws HSJException {
		sendRequest(insertPacket, false);
	}

	public Object requsetFind(IPacket findPacket) throws HSJException {
		return sendRequest(findPacket, false);
	}
	
	public Object requsetModify(IPacket modifyPacket) throws HSJException {
		return sendRequest(modifyPacket, false);
	}
	
	
	public Future<Boolean> asynRequsetInsert(final IPacket insertPacket) throws HSJException {
		Future<Boolean> future = dbOperationExecutor.submit(new Callable<Boolean>() {  
            public Boolean call() throws Exception {
        		return (Boolean) sendRequest(insertPacket, true);
            }
		});
		
		return future;
	}
	
	public Future<Object> asynRequsetFind(final IPacket findPacket) throws HSJException {
		Future<Object> future = dbOperationExecutor.submit(new Callable<Object>() {  
            public Object call() throws Exception {
        		return sendRequest(findPacket, true);
            }
		});
		
		return future;
	}
	
	public Future<Object> asynRequsetModify(final IPacket modifyPacket) throws HSJException {
		Future<Object> future = dbOperationExecutor.submit(new Callable<Object>() {  
            public Object call() throws Exception {
        		return sendRequest(modifyPacket, true);
            }
		});
		
		return future;
	}
	
	private boolean innerOpenIndex(ChannelItem channelItem, String indexId, String dbName, String tblName,
			String indexName, List<String> columnNames, List<String> filterColumnNames) throws HSJTimeoutException {
		indexIdDbNameMap.putIfAbsent(indexId, dbName);
		IPacket openIndexPacket = PacketFactory.getInstance()
				.createOpenIndexPacket(indexId, dbName, tblName, indexName,
						columnNames, filterColumnNames);
		return innerOpenIndex(channelItem, (OpenIndexPacket)openIndexPacket);
	}
	
	private boolean innerOpenIndex(ChannelItem channelItem, OpenIndexPacket openIndexPacket) throws HSJTimeoutException{
		indexIdOpenIndexMap.put(openIndexPacket.getIndexId(), (OpenIndexPacket)openIndexPacket);

		return (Boolean) innerSendRequest(channelItem, openIndexPacket, false);
	}
	
	private boolean innerAuth(ChannelItem channelItem, String authType, String authKey) throws HSJTimeoutException{
		IPacket authPacket = PacketFactory.getInstance().createAuthPacket(authKey);
		
		return (Boolean) innerSendRequest(channelItem, authPacket, false);
	}
	
	private Object sendRequest(IPacket packet, boolean isAsyn) throws HSJException{
		String dbName = indexIdDbNameMap.get(packet.getIndexId());
		ChannelItem channelItem = getChannelFromPool(dbName, packet.getIndexId());
		
		return innerSendRequest(channelItem, packet, isAsyn);
	}
	
	private Object innerSendRequest(ChannelItem channelItem, IPacket packet, boolean isAsyn) throws HSJTimeoutException{
		Channel channel = channelItem.getItem();
		
		pendingRequest.put(channel.getId(), packet);
		if(logger.isDebugEnabled()){
			logger.debug("send request, id is: " + channel.getId());
			logger.debug(packet.encode());
		}
		logger.info(packet.encode());
		channel.write(packet.encode());
		if(isAsyn){
			packet.await();
		} else {
			if(!packet.await(timeout)){
				throw new HSJTimeoutException("hs request " + timeout + " timeout!");
			}
		}

		return packet.decode();
	}
		
	public ConcurrentHashMap<Integer,IPacket> getPendingRequest(){
		return pendingRequest;
	}
	
	public void removePendingRequest(Integer channelId){
		pendingRequest.remove(channelId);
	}
	
	/**
	 * 
	 * 描述：回收使用过的channel
	 * @return
	 */
	public void addChannelPool(String indexId, Channel channel){
		if(logger.isDebugEnabled()){
			logger.debug("addChannelPool for name: " + indexId);
		}
		//auth 时，indexId为空
		if(indexId != null){
			String dbName = indexIdDbNameMap.get(indexId);
			ChannelItem channelItem = new ChannelItem(System.currentTimeMillis(), channel, true);
			if(channelPool.get(dbName) == null){//初始化
				LinkedBlockingQueue<ChannelItem> channelQueue = new LinkedBlockingQueue<ChannelItem>();
				channelPool.put(dbName, channelQueue);
				channelQueue.offer(channelItem);
			} else {
				channelPool.get(dbName).offer(channelItem);
			}
			channelItemMap.put(channelItem, true);
		}
	}
	
	public void setPoolSize(int poolSize) {
		this.poolSize = poolSize;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	private ChannelItem getChannelFromPool(final String dbName, final String indexId)
			throws HSJException {
		ChannelItem channelItem = channelPool.get(dbName).peek();
		Future<?> future = null;

		if (channelItem == null) {
			
			connLock.lock();
			try{
				if (curConnCount.get() < maxActive) {
					//first increment count
					curConnCount.incrementAndGet();
					if(logger.isDebugEnabled()){
						logger.debug("dynamic adjust conn, " + curConnCount.get());
					}
					future = connExecutor.submit(new Runnable() {
						public void run() {
							try {
								addConn(dbHostMap.get(dbName), dbName, indexId);
							} catch (HSJTimeoutException e) {
								throw new RuntimeException(e);
							}
						}
					});
				}
			}finally{
				connLock.unlock();
			}

				try {
					if (future != null && future.get() == null) {
						// if unconnected, wait 1s
						future.get(1000, TimeUnit.MILLISECONDS);
					}
				} catch (InterruptedException e) {
					throw new HSJException("get channel from pool error!", e);
				} catch (ExecutionException e) {
					throw new HSJException("thread executor error!", e);
				} catch (TimeoutException e) {
					throw new HSJException("thread timeout error!", e);
				}
		}

		try {
			channelItem = channelPool.get(dbName).take();
		} catch (InterruptedException e) {
			throw new HSJException("thread executor error!", e);
		}

		return channelItem;
	}
	
	private Channel getChannel(ChannelFuture future){
		boolean notTimeout = future.awaitUninterruptibly(2, TimeUnit.SECONDS);
		if(notTimeout){
			if(future.isSuccess()){
				return future.awaitUninterruptibly().getChannel();
			}
		}
		
		return null;
	}
}
